package com.smdp.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.smdp.dto.Result;
import com.smdp.entity.User;
import com.smdp.entity.Voucher;

import com.smdp.entity.VoucherSchedule;
import com.smdp.mapper.VoucherScheduleMapper;
import com.smdp.service.IUserService;
import com.smdp.service.IVoucherScheduleService;
import com.smdp.service.IVoucherService;
import com.smdp.utils.*;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;

import static com.smdp.utils.RabbitMQConstants.*;
import static com.smdp.utils.RedisConstants.SECKILL_SCHEDULE_LOCK_KEY;
import static com.smdp.utils.VoucherConstants.SECKILL_VOUCHER;

/**
 * 预约秒杀券服务实现类。
 * 包含预约秒杀券（用户调用）、新增秒杀券提醒（商户新增秒杀券时调用）、声明交换机、监听秒杀MQ队列、发送短信任务和发送短信的线程池。
 * 其中，秒杀券提醒逻辑为：商户新增秒杀券->新增秒杀券提醒，发送到RabbitMQ->秒杀开始，监听到RabbitMQ消息->开启独立线程->
 * 向所有预约了秒杀券的用户发送短信。
 *
 * @author 蔡方珩
 */
@Service
public class VoucherScheduleServiceImpl extends ServiceImpl<VoucherScheduleMapper, VoucherSchedule> implements IVoucherScheduleService {
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private IVoucherService voucherService;

    @Resource
    private AliyunSendSmsServiceImpl aliyunSendSmsService;

    @Resource
    private IUserService userService;

    @Resource
    private RabbitTemplate rabbitTemplate;


    /**
     * 预约秒杀券
     * @param voucherId 秒杀券id
     * @return 成功：优惠券id / 失败：失败原因
     */
    @Override
    public Result scheduleSeckillVoucher(Long voucherId) {
        // 布隆过滤器
        boolean contains = BloomFilterUtil.seckillBloomFilter.contains(voucherId);
        if (!contains){
            return Result.fail("优惠券不存在！");
        }
        // 查询数据库是否有优惠券
        Voucher voucher = voucherService.query()
                .eq("voucher_id", voucherId)
                .one();
        if (voucher == null){
            return Result.fail("优惠券不存在！");
        }
        if (Objects.equals(voucher.getType(), SECKILL_VOUCHER)){
            return Result.fail("该优惠券不是秒杀券！");
        }
        // 获取订单、用户
        Long userId = UserHolder.getUser().getId();
        User user = userService.getById(userId);
        String phone = user.getPhone();
        // 存储订阅
        VoucherSchedule voucherSchedule = new VoucherSchedule();
        voucherSchedule.setVoucherId(voucherId);
        voucherSchedule.setPhone(phone);
        voucherSchedule.setUserId(userId);
        save(voucherSchedule);
        return Result.ok(voucherId);
    }

    /**
     * 新增秒杀券时，新增秒杀开始提醒，储存在RabbitMQ
     * @param voucher 新增的秒杀券
     */
    @Override
    public void addVoucherSchedule(Voucher voucher) {
        int delayedDuration = (int) Duration.between(LocalDateTime.now(), voucher.getSeckillBeginTime()).toMillis();
        String voucherId = voucher.getId().toString();
        rabbitTemplate.convertAndSend(DELAYED_EXCHANGE, DELAYED_ROUTE_KEY, voucherId,(msg -> {
            //发送消息 并设置delayedTime
            msg.getMessageProperties().setDelay(delayedDuration);
            return msg;
        }));
    }

    /**
     * 处理秒杀提醒的线程池
     */
    private static final ThreadPoolExecutor SECKILL_NOTIFY_EXECUTOR = new ThreadPoolExecutor(
            1, //核心线程池大小
            8, //最大线程数量，获取CPU核数 System.out.println(Runtime.getRuntime().availableProcessors());
            3, //超时时间，没人调用时就会释放
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(20),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy()
    );

    /**
     * 监听秒杀提醒MQ队列。秒杀开始时，开启新线程，将优惠券信息推送给用户。
     *
     * @param message MQ消息
     * @param channel channel
     */
    @Override
    @RabbitListener(queues = DELAYED_QUEUE)
    public void handleSchedule(Message message, Channel channel) throws IOException {
        // 1.1.接收消息
        String voucherId = new String(message.getBody(), StandardCharsets.UTF_8);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        // 2.1.获取互斥锁
        String lockKey = SECKILL_SCHEDULE_LOCK_KEY + voucherId;
        SimpleRedisLock lock = new SimpleRedisLock(lockKey, stringRedisTemplate);
        boolean isLock = lock.tryLock(20);
        // 2.2.判断是否获取锁成功
        if (isLock) {
            //2.3.成功，开启独立线程，通知用户秒杀开始
            SECKILL_NOTIFY_EXECUTOR.submit(() -> {
                try {
                    // 通知用户秒杀开始
                    this.sendSeckillMsg2User(voucherId);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                } finally {
                    // 释放锁
                    lock.unlock();
                }
            });
        }
    }

    /**
     * 短信通知用户秒杀开始
     * @param voucherId 优惠券id
     */
    public void sendSeckillMsg2User(String voucherId) {
        // 查询优惠券
        Voucher voucher = voucherService.getOne(new QueryWrapper<Voucher>()
                .eq("id", voucherId));
        if (voucher == null){
            return;
        }
        // 查询订阅了优惠券的用户
        List<VoucherSchedule> list = query()
                .select("phone")
                .eq("voucher_id", voucherId)
                .list();
        if (list == null || list.isEmpty()) {
            return;
        }
        // 推送给用户
        String message = voucher.getTitle() + "优惠券开始秒杀了！登录水木点评抢购吧";
        for (VoucherSchedule voucherSchedule: list) {
            String phone = voucherSchedule.getPhone();
            // 发送短信通知用户
            String s = aliyunSendSmsService.sendMsgToUser(phone, message);
            log.trace(s);
        }
        // 删除优惠券对应的订阅信息
        remove(new QueryWrapper<VoucherSchedule>()
                .eq("voucher_id", voucherId));
    }



}
